#pragma warning(disable : 4355)
#include "Avalon/IO/DelimitedReader.hpp"
#include <climits>
#include <boost/bind.hpp>
#include "Avalon/IO/Buffer.hpp"

using namespace Avalon;
using namespace Avalon::IO;
using namespace Avalon::Threading;
using namespace boost;
using namespace std;

namespace {
  const int INITIAL_READ_CHUNK_SIZE = 32;
}

DelimitedReader::DelimitedReader(Carry<Reader> source, char delimiter)
    : m_source(source),
      m_delimiter(delimiter),
      m_writer(&m_reader),
      m_readChunkSize(INITIAL_READ_CHUNK_SIZE) {
  Async<int>::State state;
  m_sizeRead.SetFinishedSlot(bind(&DelimitedReader::OnRead, this),
    Store(state));
}

DelimitedReader::~DelimitedReader() {}

void DelimitedReader::Read(Buffer* destination, Out<Async<int> > sizeRead) {
  ReadSource();
  m_reader.Read(destination, Store(sizeRead));
}

void DelimitedReader::Read(void* destination, int size,
    Out<Async<int> > sizeRead) {
  ReadSource();
  m_reader.Read(destination, size, Store(sizeRead));
}

void DelimitedReader::Read(Buffer* destination, int size,
    Out<Async<int> > sizeRead) {
  ReadSource();
  m_reader.Read(destination, size, Store(sizeRead));
}

void DelimitedReader::ReadSource() {
  m_sizeRead.ResetResult();
  m_destination.Reset();
  m_source->Read(&m_destination, m_readChunkSize, Store(m_sizeRead));
}

void DelimitedReader::OnRead() {
  if(m_sizeRead.GetState() == Async<int>::EXCEPTION) {
    m_reader.Break(m_sizeRead.GetException());
    return;
  }
  int sizeRead = m_sizeRead.Get();
  vector<int> delimiterIndicies;
  for(int i = 0; i < sizeRead; ++i) {
    if(m_destination.GetData()[i] == m_delimiter) {
      delimiterIndicies.push_back(i);
    }
  }
  if(delimiterIndicies.empty()) {
    m_remainder.Append(m_destination);
    if(sizeRead == m_readChunkSize) {
      m_readChunkSize *= 2;
    }
    ReadSource();
    return;
  }
  Buffer firstMessage = m_remainder;
  firstMessage.Append(m_destination.GetData(), delimiterIndicies.front());
  m_remainder.Reset();
  Buffer remainingMessages = m_destination;
  m_destination.Reset();
  m_remainder.Append(remainingMessages.GetData() + delimiterIndicies.back(),
    remainingMessages.GetSize() - delimiterIndicies.back() - 1);
  Async<void> writeResult;
  m_writer.Write(firstMessage, Store(writeResult));
  for(vector<int>::const_iterator i = delimiterIndicies.begin() + 1;
      i != delimiterIndicies.end(); ++i) {
    writeResult.Reset();
    m_writer.Write(remainingMessages.GetData() + *(i - 1) + 1,
      *i - (*(i - 1) + 1), Store(writeResult));
  }
}
